// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin
#pragma once

#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "coin2/base/config/app_config.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/factory/feed/raw_processor.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/flow/reader.h"
#include "coin2/flow/topic_string.h"
#include "coin2/flow/topic_subscription.h"
#include "orio/io/base.h"
#include "orio/record/topic.h"

namespace coin2::feed::archive {

namespace impl {
using coin2::base::config::FromIntervalConfig;
using coin2::base::config::ValidateArchiveConfig;
using coin2::exchange::base::executor::ArchiveSyncExecutor;
using coin2::exchange::base::feed::FeedArchiveMixin;
using coin2::exchange::base::feed::IFeedSubsystem;
using coin2::exchange::base::feed::TopicRegister;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::feed::FeedSystemConfig;
using coin2::exchange::feed::RawParserConfig;

class RawArchiveFeedSystem : public IFeedSubsystem, public FeedArchiveMixin {
 public:
  RawArchiveFeedSystem(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const coin2::exchange::di::ConstantInjection& ci,
      const FeedSystemConfig& config,
      ArchiveSyncExecutor* ass,
      exchange::base::feed::IFeedSubscriber* callback);

  void Prepare() override {}
  void Cleanup() override {}

  // The timestamp of next packet
  // if eof, return 0.
  // you should not call this again after EOF
  int64_t PeekTimestamp() override;
  void ConsumeRecord() override;

  std::vector<IProduct*> GetProducts() const override {
    std::vector<IProduct*> out;
    for (const auto& p : products_) {
      out.push_back(p.get());
    }
    return out;
  }

 protected:
  RawParserConfig BuildRawParserConfigForRecipe(
      const std::string& recipe,
      const MarketExchangeApi& mea,
      const coin2::exchange::di::ConstantInjection& ci);
  const FeedSystemConfig& config_;

  int64_t interval_first_;
  // states
  bool has_packet_ = false;
  exchange::base::executor::TopicRecordPacket packet_;
  std::unique_ptr<flow::FlowReader> reader_;
  std::vector<std::pair<std::string, std::unique_ptr<
      exchange::base::feed::FeedParsingProcessor>>> processors_;
  std::vector<std::unique_ptr<IProduct>> products_;
  TopicRegister topic_register_;
};

}  // namespace impl

using impl::RawArchiveFeedSystem;

}  // namespace coin2::feed::archive
